package com.distributed.master;

import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.retry.ExponentialBackoffRetry;

import java.io.*;
import java.net.Socket;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;

public class ZookeeperConnector extends Thread {
    // TODO: fill server_proxy
    private static final String SERVER_PROXY = "localhost:2181";
    private static final int FAILURE_PORT = 8010;

    public ZookeeperConnector(HashMap<String, List<String>> directory) throws IOException {
        try {
            getNodesInfo(directory);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Override
    public void run() {
        super.start();
        try {
            getNodesInfo(MasterServer.directory);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /*
     * Author: Xinbei Wu
     * Name: getNode
     * Function: get and update region servers information
     * Params: directory(HashMap<> | master's directory)
     * */
    public static void getNodesInfo(HashMap<String, List<String>> directory) throws Exception {
        // set retry policy
        RetryPolicy retry_policy = new ExponentialBackoffRetry(2000, 3);
        CuratorFramework client = CuratorFrameworkFactory.builder()
                .connectString(SERVER_PROXY)
                .retryPolicy(retry_policy)
                .namespace("service")
                .build();
        client.start();

        PathChildrenCache cache = new PathChildrenCache(client, "/", true);
        cache.start();
        // add listener
        cache.getListenable().addListener((c, event) -> {
            System.out.println(event.getType());
            // get all nodes under path “/”

            if (event.getType() == PathChildrenCacheEvent.Type.CHILD_REMOVED) { // 有节点被移除
                List<String> child_nodes = client.getChildren().forPath("/");
                String path = event.getData().getPath();
                System.out.println("[INFO] " + path + " disconnects.");

                String url = new String("-1");
                // find an empty node
                for (String node : child_nodes) {
                    String tables = new String(client.getData().forPath("/" + node));
                    String[] info = tables.split(" ");

                    if (info.length == 1) {
                        url = info[0];
                        break;
                    }
                }
                // send message
                if (url.equals("-1")) {
                    System.out.println("[ERROR] Cannot find empty region node.");
                } else {
                    Socket socket = new Socket(url, FAILURE_PORT);
                    DataOutputStream output_stream = new DataOutputStream((OutputStream) socket.getOutputStream());
                    output_stream.writeUTF(path);
                    DataInputStream input_stream = new DataInputStream((InputStream) socket.getInputStream());
                    String res = input_stream.readUTF();
                    System.out.println(res);

                    socket.close();
                }
            }

            // update master directoryC:\\Users\\10721\\Desktop\\
            List<String> child_nodes = client.getChildren().forPath("/");
            directory.clear();
            for (String node : child_nodes) {
                String tables = new String(client.getData().forPath("/" + node));
                String[] info = tables.split(" ");

                String url = info[0];
                ArrayList<String> table_names = new ArrayList<>();
                if (info.length > 1) {
                    table_names.addAll(Arrays.asList(info).subList(1, info.length));
                }
                directory.put(url, table_names);
            }
            // prompt
            System.out.println("Relationship between Region & tables is shown below: ");
            System.out.println(directory);
        });

        Thread.sleep(Integer.MAX_VALUE);
    }
}
